package app.htby.utils;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

@Component
public class KafkaProduceUtils {

	/** 
	 * kafka生产者配置
	 * 
	 * 需要注意的有两点:
	 * 1、acks回令。如果必须等待回令，那么设置acks为all；否则，设置为-1；等待回令会有性能损耗。
	 * 2、生产者在发送消息的过程中，会自己默认批量提交。所以，如果单条指令的发送请求，记得发送完后flush才能生效。
	 * 3、SimplePartitioner2.java为kafaka分区，可选项。
	 *  
	 **/
	/** 设置生产者需要连接的kafka地址 */
	public static String bootstrapServers;
	/** 回令类型 */
	public static String acks;
	/** 重试次数 */
	public static String retries;
	/** 批量提交大小 */
	public static String batch_size;
	/** 提交延迟等待时间（等待时间内可以追加提交） */
	public static String linger_ms;
	/** 缓存大小 */
	public static String buffer_memory;
	/** 序列化方法 */
	public static String key_serializer;
	public static String value_serializer;
	
	/** topic */
	public static String hearbeatTopic;
	/** key */
	public static String hearbeatKey;
	
	@Value("${kafka.hearbeat.key}")
	public void setHearbeatKey(String hearbeatKey) {
		KafkaProduceUtils.hearbeatKey = hearbeatKey;
	}
	
	@Value("${kafka.hearbeat.topic}")
	public void setHearbeatTopic(String hearbeatTopic) {
		KafkaProduceUtils.hearbeatTopic = hearbeatTopic;
	}

	@Value("${kafka.produce.value.serializer}")
	public void setValueSerializer(String value_serializer) {
		KafkaProduceUtils.value_serializer = value_serializer;
	}

	@Value("${kafka.produce.key.serializer}")
	public void setKeySerializer(String key_serializer) {
		KafkaProduceUtils.key_serializer = key_serializer;
	}

	@Value("${kafka.produce.buffer.memory}")
	public void setBufferMemory(String buffer_memory) {
		KafkaProduceUtils.buffer_memory = buffer_memory;
	}

	@Value("${kafka.produce.linger.ms}")
	public void setLingerMs(String linger_ms) {
		KafkaProduceUtils.linger_ms = linger_ms;
	}

	@Value("${kafka.produce.batch.size}")
	public void setBatchSize(String batch_size) {
		KafkaProduceUtils.batch_size = batch_size;
	}

	@Value("${kafka.produce.retries}")
	public void setRetries(String retries) {
		KafkaProduceUtils.retries = retries;
	}

	@Value("${kafka.produce.acks}")
	public void setAcks(String acks) {
		KafkaProduceUtils.acks = acks;
	}

	@Value("${kafka.produce.bootstrap.servers}")
	public void setBootstrapServers(String bootstrapServers) {
		KafkaProduceUtils.bootstrapServers = bootstrapServers;
	}
}
